package com.atguigu.sparkstreaming.demos

import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.rdd.RDD
import org.apache.spark.streaming.dstream.DStream
import org.apache.spark.streaming.kafka010.{ConsumerStrategies, HasOffsetRanges, KafkaUtils, LocationStrategies, OffsetRange}
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
 * Created by Smexy on 2022/5/23
 *
 *    获取offsets：  val offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
 *
 *    DS和RDD的关系:
 *          DS是 RDD的无限集合。
 *          在Sparkstreaming中，DS是按照批次来进行计算。
 *          对DS的各种转换，其实是对DS中封装的每个批次的RDD的转换运算。
 *
 *    转换运算:
 *            某些转换算子，在DS中定义了，在RDD中也定义了。例如: map,filter,flatmap
 *                    DS.map
 *            某些转换算子，在DS中没有定义，在RDD中定义了。例如: sortByKey, 如何操作?
 *
 *     此时需要将对DS的运算，转换为对DS中RDD的运算:
 *            transform:  transform[U: ClassTag](transformFunc: RDD[T] => RDD[U]): DStream[U]
 *
 *            foreachRDD:  foreachRDD(foreachFunc: RDD[T] => Unit): Unit。 输出算子。
 *
 *            根据是否是转换后再转换，用transform，转换后就直接输出，用foreachRDD
 *
 *
 */
object GetOffsetsDemo1 {

  def main(args: Array[String]): Unit = {

    val streamingContext = new StreamingContext("local[*]", "SparkStreamingKafkaDemo", Seconds(5))


    val kafkaParams = Map[String, Object](
      "bootstrap.servers" -> "hadoop102:9092,hadoop103:9092",
      "key.deserializer" -> classOf[StringDeserializer],
      "value.deserializer" -> classOf[StringDeserializer],
      "group.id" -> "atguigu1227",
      "auto.offset.reset" -> "latest",
      // ①取消offsets的自动提交
      "enable.auto.commit" -> "false"
    )

    /*
        消费的主题。
            通常一个 DStream 只消费一个主题。
            如果消费多个主题，一般是创建多个DStream
     */
    val topics1 = Array("topicA")

    /*
        使用kafka提供的工具类，创建DStream(流，代表源源不断地去消费主题中的数据，所组成的无限集合)
     */
    val stream = KafkaUtils.createDirectStream[String, String](
      streamingContext,
      LocationStrategies.PreferConsistent,
      ConsumerStrategies.Subscribe[String, String](topics1, kafkaParams)
    )

    val ds1: DStream[String] = stream.transform(rdd => {

      val offsetRanges: Array[OffsetRange] = rdd.asInstanceOf[HasOffsetRanges].offsetRanges

      offsetRanges.foreach(offsetRange => println(offsetRange))

      rdd.map(record => record.value())

    })

    ds1.print(1000)

    /*stream.foreachRDD(rdd => {

      /*
           获取到的当前批次数据的偏移量

           OffsetRange: 一个分区的偏移量信息。

           OffsetRange(topic: 'topicA', partition: 2, range: [3 -> 3])
              关注:  untilOffset，才是要提交的位置

       */

      val offsetRanges: Array[OffsetRange] = rdd.asInstanceOf[HasOffsetRanges].offsetRanges

      offsetRanges.foreach(offsetRange => println(offsetRange))

    })*/






    // 输出结果
   // ds.print(1000)

    //   ②在输出结果后，再手动提交offsets
    // xxxxxx

    // 启动app
    streamingContext.start()

    // 阻塞当前线程，让程序24h不停运行
    streamingContext.awaitTermination()


  }

}
